package com.chengcang.admin.modules.log.binlog;

import cn.hutool.core.collection.CollectionUtil;
import com.chengcang.admin.modules.log.binlog.listener.MysqlBinLogListener;
import com.chengcang.admin.modules.log.binlog.model.BinLogConstants;
import com.chengcang.admin.modules.log.binlog.model.BinLogItem;
import com.chengcang.admin.modules.log.binlog.model.BinlogConfig;
import com.chengcang.admin.modules.log.entity.SysLogOperateEntity;
import com.chengcang.admin.modules.log.service.SysLogOperateService;
import com.chengcang.admin.modules.sys.dao.DdlBaseMapper;
import com.chengcang.admin.modules.sys.dao.SysUserDao;
import com.chengcang.admin.modules.sys.entity.SysUserEntity;
import com.chengcang.common.utils.JsonUtils;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author fankunfeng
 * @date 2023-01-31 10:53:54
 **/
@Slf4j
@Component
@Order(value = 1)
public class TourBinLogListener implements CommandLineRunner {

    @Value("${binlog.exec}")
    private Boolean exec;
    @Resource
    private BinLogConstants binLogConstants;
    @Resource
    private DdlBaseMapper ddlBaseMapper;
    @Resource
    private SysLogOperateService sysLogOperateService;
    @Resource
    private SysUserDao sysUserDao;

    private static Cache<Long, SysUserEntity> cache = CacheBuilder.newBuilder().maximumSize(100).expireAfterWrite(5, TimeUnit.MINUTES).build();
    private static BlockingQueue<SysLogOperateEntity> blockingQueue = new LinkedBlockingQueue<>();

    @Override
    public void run(String... args) throws Exception {
        log.info("binlog日志解析是否开启：{}", exec);
        log.info("初始化配置信息：{}", binLogConstants.toString());
        if (exec) {
            binlogExec();
        }
    }

    private void binlogExec() throws IOException, InterruptedException {
        // 初始化配置信息
        BinlogConfig conf = new BinlogConfig(binLogConstants.getHost(), binLogConstants.getPort(), binLogConstants.getUsername(), binLogConstants.getPasswd());

        // 初始化监听器
        BinaryLogClient client = new BinaryLogClient(conf.getHost(), conf.getPort(), conf.getUsername(), conf.getPasswd());
        EventDeserializer eventDeserializer = new EventDeserializer();
//        eventDeserializer.setCompatibilityMode(//序列化
//                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
//                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
//        );
        client.setEventDeserializer(eventDeserializer);

        MysqlBinLogListener mysqlBinLogListener = new MysqlBinLogListener(client, ddlBaseMapper);

        // 获取table集合
        List<String> tableList = BinLogUtils.getListByStr(binLogConstants.getTable());
        if (CollectionUtil.isEmpty(tableList)) {
            return;
        }
        // 注册监听
        tableList.forEach(table -> {
//            log.info("注册监听信息，注册DB：" + binLogConstants.getDb() + "，注册表：" + table);
            try {
                mysqlBinLogListener.regListener(binLogConstants.getDb(), table, item -> {
//                    log.info(" 监听逻辑处理 item: table {}  \nitem before :{} \n item after:{} ", table, item.getBefore(), item.getAfter());
                    SysLogOperateEntity sysLogOperateEntity = operationLogInsert(table, item);
                    blockingQueue.add(sysLogOperateEntity);
                });
            } catch (Exception e) {
                log.error("BinLog监听异常：" + e);
            }
        });
        // 多线程消费
        mysqlBinLogListener.parse();

        client.connect();

        Executors.newSingleThreadExecutor().execute(new InsertBatch(sysLogOperateService));
    }

    private static class InsertBatch extends Thread {

        private SysLogOperateService sysLogOperateService;
        private List<SysLogOperateEntity> list;


        public InsertBatch(SysLogOperateService sysLogOperateService) {
            this.sysLogOperateService = sysLogOperateService;
            list = new ArrayList<>(20);
        }

        @Override
        public void run() {
            try {
                loop();
            } catch (Exception e) {
                log.error("插入消息异常：{}", e);
            }
        }

        private void loop() {
            // 20条插入一次
            while (true) {
//                log.info("执行尝试插入数据！");
                try {
                    SysLogOperateEntity logOperateEntity = blockingQueue.take();
                    if (list.size() == 20) {
                        sysLogOperateService.insertBatch(list);
                        list.clear();
                    } else {
                        list.add(logOperateEntity);
                    }
                    TimeUnit.MICROSECONDS.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private SysLogOperateEntity operationLogInsert(String table, BinLogItem item) {
        SysLogOperateEntity operateEntity = new SysLogOperateEntity();
        operateEntity.setTableName(table);
        operateEntity.setBeforeInfo(JsonUtils.toJsonString(item.getBefore()));
        operateEntity.setAfterInfo(JsonUtils.toJsonString(item.getAfter()));
        Map<String, Serializable> afterMap = item.getAfter();

        operateEntity.setOperation(BinLogUtils.getOptType(item));

        Serializable updateDate = afterMap.get("update_date");
        Serializable createDate = afterMap.get("create_date");
        Serializable creator = afterMap.get("creator");
        Serializable updater = afterMap.get("updater");
        Serializable tenantId = afterMap.get("tenant_id");

        if (updateDate instanceof Date) {
            operateEntity.setUpdateDate((Date) updateDate);
            operateEntity.setOperationTime((Date) updateDate);
        }
        if (createDate instanceof Date) {
            operateEntity.setCreateDate((Date) createDate);
        }
        if (creator instanceof Long) {
            operateEntity.setCreator((Long) creator);
        }
        if (updater instanceof Long) {
            operateEntity.setUpdater((Long) updater);
            operateEntity.setOperatorId((Long) updater);
            SysUserEntity sysUserEntity = cache.getIfPresent(operateEntity.getOperatorId());
            if (sysUserEntity == null) {
                sysUserEntity = sysUserDao.getById(operateEntity.getOperatorId());
                cache.put(operateEntity.getOperatorId(), sysUserEntity);
            }
            operateEntity.setOperatorName(sysUserEntity.getRealName());
        }
        Optional.ofNullable(tenantId).ifPresent(o -> operateEntity.setTenantId((Long) tenantId));
        return operateEntity;
    }

}
